package com.example.mq.mqsever.datacenter;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: WHY
 * Date: 2023-08-28
 * Time: 14:37
 */

import com.example.mq.common.BinaryTool;
import com.example.mq.common.MqException;
import com.example.mq.mqsever.core.MSGQueue;
import com.example.mq.mqsever.core.Message;


import java.awt.datatransfer.FlavorEvent;
import java.io.*;
import java.util.LinkedList;
import java.util.Random;
import java.util.Scanner;

/**
 * 这个类,来针对硬盘上的消息进行管理
 */
public class MessageFileManager {
    //定义一个内部类,表示该对队列的统计信息
    //优先考虑使用static,表明是静态内部类
    static public class Stat {
        public int totalCount;//总消息数据
        public int validCount;//有效消息数据
    }
    public  void init(){
        //暂时不需要做啥额外的初始工作,后续扩展
    }

    //约定消息队列所在的目录和文件名
    //这个方法用来获取指定队列对应的消息文件所在路径
    public String getQueueDir(String queueName) {
        return "./data/" + queueName;
    }

    //这个方法用来获取该队列的消息数据文件路径(也就是文件内容),这个文件是一个二进制格式的文件
    //注意,二进制文件,使用txt来作为后缀,不太好.这个一般用来表示文本
    //可以用.bin 或者是.dat来作为后缀
    private String getQueueDataPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data.bin";
    }

    //这个方法用来获取该队列的消息统计文件路径(放统计信息)(这是一个文本文件)
    private String getQueueStatPath(String queueName) {
        return getQueueDir(queueName) + "/queue_stat.txt";
    }
//读有效消息
    private Stat readStat(String queueName) {
        //由于当前的消息统计文件是文本文件,可以直接使用Scanner来读取文件内容
        Stat stat = new Stat();
        try (InputStream inputStream = new FileInputStream(getQueueStatPath(queueName))) {
            Scanner scanner = new Scanner(inputStream);
            stat.totalCount = scanner.nextInt();//读到的第一个就是总的数据个数
            stat.validCount = scanner.nextInt();//第二个就是有效的数据个数
            return stat;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }
//写有效消息
    private void writeStat(String queueName, Stat stat) {
        //使用PrinterWriter来写文件,
        //OutPutStream打开文件,默认情况下会直接把原文件清空,相当于新的数据覆盖了旧的数据
        try (OutputStream outputStream = new FileOutputStream(getQueueStatPath(queueName))) {
            PrintWriter printWriter = new PrintWriter(outputStream);
            printWriter.write(stat.totalCount + "\t" + stat.validCount);
            printWriter.flush();//数据从缓冲区到硬盘

        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    //创建队列对应的文件和目录
    public void createQueueFiles(String queueName) throws IOException {
        //1.根据队列创建目录
        File baseDir = new File(getQueueDir(queueName));
        if (!baseDir.exists()) {
            //不存在,就创建这个目录
            //这个方法可以创建多级目录

            boolean ok = baseDir.mkdirs();
            if (!ok) {
                throw new IOException("创建目录失败! baseDir=" + baseDir.getAbsolutePath());
            }
        }


        //创建队列数据文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()) {
            boolean ok = queueDataFile.createNewFile();
            if (!ok) {
                throw new IOException("创建文件失败! queueDataFile" + queueDataFile.getAbsolutePath());
            }
        }
        //创建数据统计文件
        File queueStatFile = new File(getQueueStatPath(queueName));
        if (!queueStatFile.exists()) {
            boolean ok = queueStatFile.createNewFile();
            if (!ok) {
                throw new IOException("创建文件失败 queueStatFile" + queueStatFile.getAbsolutePath());
            }
        }
        //给消息统计文件设置初值   0\t0
        Stat stat = new Stat();
        stat.totalCount = 0;
        stat.validCount = 0;
        writeStat(queueName, stat);
    }

    //删除队列的目录和文件
    //队列也是可以被删除的,当队列删除以后,对应的消息文件,也就随之删除了
    public void destroyQueueFiles(String queueName) throws IOException {
        //先删除文件,再删除目录
        File queueDataFile = new File(getQueueDataPath(queueName));
        boolean ok1 = queueDataFile.delete();
        File queueStatFile = new File(getQueueStatPath(queueName));
        boolean ok2 = queueStatFile.delete();
        File baseDir = new File(getQueueDir(queueName));
        boolean ok3 = baseDir.delete();
        if (!ok1 || !ok2 || !ok3) {
            //任意一个删除失败,整体就算删除失败
            throw new IOException("删除队列目录和文件失败 baseDir" + baseDir.getAbsolutePath());
        }
    }

    //检查队列的文件和目录是否存在
    //后续有生产者的broker server生产消息了,这个消息可能就需要记录到文件上(取决于消息是否要持久化,如果持久化,就需要写到文件,那么得确保文件是存在的,不然那写了个寂寞)
    public boolean checkFilesExists(String queueName) {
        //判定队列的数据文件和统计文件是否都存在

        File queueDataFile = new File(getQueueDataPath(queueName));
        if (!queueDataFile.exists()) {
            return false;
        }
        File queueStatFile = new File(getQueueStatPath(queueName));
        if (!queueStatFile.exists()) {
            return false;
        }
        return true;
    }

    //这个方法用来把一个新的消息放到队列对应的文件中
    //queue 表示要把消息写入的队列,message是要写的消息
    public void sendMessage(MSGQueue queue, Message message) throws MqException, IOException {
        //1.检查一下当前要写入的队列对应的文件是否存在
        if (!checkFilesExists(queue.getName())) {
            throw new MqException("[MessageFileManager]队列对应的文件不存在 queueName" + queue.getName());

        }
        //2.写文件之前,把对象进行序列化转为
        //二进制字节数组
        byte[] messageBinary = BinaryTool.toBytes(message);
        //3.我们在文件前要确定消息的
        //起始位置和末位置
        //新的Message数据写入到对垒数据文件的末尾,此时Message对象的offsetBeg=当前文件长度+4,
        //offsetEnd就是当前文件长度+4+message自身长度
        synchronized (queue) {
            File queueDataFile = new File(getQueueDataPath(queue.getName()));
            //通过这个方法,queueDataFile.length()就会获取到文件的长度,单位是字节
            message.setOffsetBeg(queueDataFile.length() + 4);
            message.setOffsetEnd(queueDataFile.length() + 4 + messageBinary.length);

            //4.写入消息到数据文件,注意,是追加到数据文件末尾,不是覆盖
            try (OutputStream outputStream = new FileOutputStream(queueDataFile, true)) {
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    //接下来要写消息的长度
                    dataOutputStream.writeInt(messageBinary.length);
                    //写入消息本体
                    dataOutputStream.write(messageBinary);


                }
            }
            //5.更新消息统计文件
            Stat stat = readStat(queue.getName());
            stat.totalCount += 1;
            stat.validCount += 1;
            writeStat(queue.getName(), stat);
        }

    }

    //删除队列的消息
    //这里的删除是逻辑删除,就是把硬盘上存储的数据里面的isValid设置为0
    public void deleteMessage(MSGQueue queue, Message message) throws IOException, ClassNotFoundException {
        //1.先把文件中的要删除的这一段数据读出来,
        //这里的读是随机访问,指定位置读写
        //FileInputStream和FileOutPutStream是从文件头开始读的,所以不用这个
        //我们用类RandomAccessFile
        synchronized (queue) {
            try (RandomAccessFile randomAccessFile = new RandomAccessFile(getQueueDataPath(queue.getName()), "rw")) {
                byte[] bufferSrc = new byte[(int) (message.getOffsetEnd() - message.getOffsetBeg())];
                //seek是移动文件的光标的

                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.read(bufferSrc);
                // 2.还原为Message对象
                Message diskMessage = (Message) BinaryTool.fromBytes(bufferSrc);
                //3..把isValid写为0
                diskMessage.setIsValid((byte) 0X0);
                //4.把上述数据重新写回到文件,此处的message对象也必须保证包含有效的offsetbeg和offsetend
                byte[] bufferDest = BinaryTool.toBytes(diskMessage);
                //虽然上面已经seek过了,但是上面seek以后,进行了读操作,就会导致光标移动,所以需要重新seek
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.write(bufferDest);
                //通过上述的方法,对于文件来说,只有一个字节发生了变化,也就是那个isValid
                //记得统计文件,把一个消息设为无效了,那么有效消息需要-1
                Stat stat = readStat(queue.getName());
                if (stat.validCount > 0) {
                    stat.validCount -= 1;
                }
                writeStat(queue.getName(), stat);
            }
        }
    }

    //使用这个方法从文件中读取出所有的消息内容,加载到内存中(具体说是放到一个链表里)
    //这个方法准备在程序启动的时候,进行调用
    //这里使用LinkedList,主要目的是为了后续进行头删操作
    //这个方法的参数,只是一个queueName而不是一个MSGQueue对象,因为这个方法不需要加锁,
    //该方法是在程序启动的时候调用,此时服务器还没有处理请求,不存在多线程调用
    public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws IOException, MqException, ClassNotFoundException {
        LinkedList<Message> messages = new LinkedList<>();
        try (InputStream inputStream = new FileInputStream(getQueueDataPath(queueName))) {
            try (DataInputStream dataInputStream = new DataInputStream(inputStream)) {
                //这个变量记录当前文件光标
                long currentOffset = 0;
                //一个文件包含了很多消息,势必要循环读取
                while (true) {
                    //读取文件长度
                    //读当前消息长度,这里的读可能会读到文件末尾,会抛出EOFException异常,和之前的很多流对象不一样

                    int messageSize = dataInputStream.readInt();
                    //按照这个长度读取消息内容
                    byte[] buffer = new byte[messageSize];
                    int actualSize = dataInputStream.read(buffer);//读满这个数组
                    if (messageSize != actualSize) {
                        //不匹配,文件有问题,格式错乱了,文件剩余空间不够了,会有这样的情况
                        throw new MqException("[MessageFileManager] 文件格式异常 queueName" + queueName);
                    }
                    //3.把这个读到的数据二进制反序列化为Message对象
                    Message message = (Message) BinaryTool.fromBytes(buffer);
//判定一下消息是否是无效的
                    if (message.getIsValid() != 0x1) {
                        //无效数据,直接跳过
                        //虽然消息无效,但是offset不要忘记更新
                        currentOffset += 4;
                        continue;
                    }
                    //5.有效数据,需要把这个Messaeg对象加入到链表中,加入之前还要填写offsetBeg和OffsetEnd
                    //计算offset的时候,需要知道当前文件光标的位置,但是这里的DataInPutStream不能提供知道光标的方法
                    //所以我们手动计算
                    message.setOffsetBeg(currentOffset + 4);
                    message.setOffsetEnd(currentOffset + 4 + messageSize);
                    currentOffset += (4 + messageSize);
                    messages.add(message);

                }
            } catch (EOFException e) {
                //这个catch不是真正的用来处理异常的,而是处理正常的业务逻辑,文件读到末尾,会被readint抛出异常
                //这个catch语句也不用做啥特殊的事情
                System.out.println("[MessageFileManager] 恢复数据完成");

            }


        }
        return messages;
    }

    //检查当前是否要针对该队列的消息数据文件进行GC
    public boolean checkGC(String queueName) {
        //判定是否要GC,是根据总消息和有效消息总数,这两个值是在消息统计文件中的
        Stat stat = readStat(queueName);
        if (stat.totalCount > 2000 && (double) stat.validCount / (double) stat.totalCount < 0.5) {
            return true;
        }
        return false;
    }
    private String getQueueDataNewPath(String queueName){
        return getQueueDir(queueName)+"/queue_data_new.bin";
    }
    //通过这个方法,真正执行垃圾回收算法
    //使用复制算法执行
    //创建一个新的文件,就是上面那个方法那个文件
    //把之前消息数据文件的有效消息都读出来,写到新的文件中
    //删除旧的文件,再把新的文件名改回queue_data.bin,
    //同时更新消息统计文件
    public void gc(MSGQueue queue) throws MqException, IOException, ClassNotFoundException {
        //进行gc的时候,是针对消息数据文件进行大洗牌,在回收过程中,其他线程不能对该队列的 消息进行修改
        //加锁,保证线程安全
        synchronized (queue){
            //由于gc操作可能会耗时,所以这里统计一下执行消耗的时间
            long gcBeg=System.currentTimeMillis();
            //1.创建一个新的文件
            File queueDataNewFile=new File(getQueueDataNewPath(queue.getName()));
            if(queueDataNewFile.exists()){
                //正常情况下,不应该存在,要是存在,说明上次gc了一半,程序意外崩溃了
                throw new MqException("[MessageFileManager] gc时发现该队列的queue_data_new已经存在! queueName"
                        +queue.getName());
            }
            boolean ok=queueDataNewFile.createNewFile();
            if(!ok){
                throw new MqException("[MessageFileManager] 创建文件失败! queueDataFile="+
                        queueDataNewFile.getAbsolutePath());
            }
        //从旧的文件中,读取出所有的有效消息对象(刚写过了,直接调用)
            LinkedList<Message> messages=loadAllMessageFromQueue(queue.getName());

            //3.把有效消息,写入到新的文件中(复制过程)
            try(OutputStream outputStream=new FileOutputStream(queueDataNewFile)){
                try(DataOutputStream dataOutputStream=new DataOutputStream(outputStream)){
                    for(Message message:messages){
                        byte[]  buffer=BinaryTool.toBytes(message);
                        //先写四个字节消息的长度
                        dataOutputStream.writeInt(buffer.length);
                        //在写文件的内容
                        dataOutputStream.write(buffer);
                    }
                }
            }
        //4.删除旧文件,新文件重命名
        File queueDataOldFile=new File(getQueueDataPath(queue.getName()));
         ok=queueDataOldFile.delete();
         if(!ok){
             //比如删的文件没有权限
             throw new MqException("[MessageFileManager] 删除旧的数据文件失败! queueDataFile="+queueDataOldFile.getAbsolutePath());
         }
         //把queue_data_new.bin 命名为queue_data.bin
      ok=  queueDataNewFile.renameTo(queueDataOldFile);
         if(!ok){
             throw new MqException("[MessageFileManager] 文件重命名失败 queueDataNewFile="+queueDataNewFile.getAbsolutePath()
             +", queueDataOleFile="+queueDataOldFile.getAbsolutePath());

         }
         //5更新统计文件
            Stat stat=readStat(queue.getName());
         stat.totalCount=messages.size();
         stat.validCount=messages.size();
         writeStat(queue.getName(),stat);





            long gcEnd=System.currentTimeMillis();
            System.out.println("[MessageFileManager] gc执行完毕! queueName"+queue.getName()+", time=" +
                    (gcEnd-gcBeg) +"ms"   );
        }
    }

}





